package com.shop.cloud.live.admin.socket;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.shop.cloud.common.core.util.R;
import com.shop.cloud.common.data.tenant.TenantContextHolder;
import com.shop.cloud.live.admin.service.LiveRoomService;
import com.shop.cloud.live.admin.service.LiveRoomSpuService;
import com.shop.cloud.live.common.constant.LiveConstant;
import com.shop.cloud.live.common.constant.LiveRoomStatusType;
import com.shop.cloud.live.common.constant.MsgCode;
import com.shop.cloud.live.common.constant.MsgType;
import com.shop.cloud.live.common.dto.LiveMessageDTO;
import com.shop.cloud.live.common.entity.LiveRoom;
import com.shop.cloud.live.common.entity.LiveRoomSpu;
import com.shop.cloud.live.common.vo.LiveMsgItemVO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 主播
 *
 * @author
 */
@ServerEndpoint("/ws/anchor/{tenantId}/{userId}/{roomId}")
@Component
public class LiveAnchorWebSocketServer {

    public static final Logger logger = LoggerFactory.getLogger(LiveAnchorWebSocketServer.class);

    public static final ConcurrentHashMap<String, Session> LIVE_ANCHOR_MAP = new ConcurrentHashMap<>(LiveConstant.DEFAULT_MAP_SIZE);

    private static LiveRoomService liveRoomService;

    private static LiveRoomSpuService liveRoomSpuService;

    @Autowired
    public void setLiveRoomService(LiveRoomService liveRoomService) {
        LiveAnchorWebSocketServer.liveRoomService = liveRoomService;
    }

    @Autowired
    public void setLiveRoomSpuService(LiveRoomSpuService liveRoomSpuService) {
        LiveAnchorWebSocketServer.liveRoomSpuService = liveRoomSpuService;
    }

    @OnOpen
    public void onOpen(Session session, @PathParam("tenantId") String tenantId, @PathParam(value = "userId") String userId, @PathParam("roomId") String roomId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        logger.info("创建主播直播连接");
        // 检查直播间
        LiveRoom liveRoom = checkLiveRoom(session, roomId, userId);
        if (liveRoom == null) {
            return;
        }
        if (Objects.equals(liveRoom.getLiveStatus(), LiveRoomStatusType.FINISHED.value())) {
            logger.info("连接一个结束的直播间，返回" + roomId);
            return;
        }

        // 已经有连接
        Session oldSession = LIVE_ANCHOR_MAP.get(LiveConstant.LIVE_ANCHOR_PREFIX + roomId + "_" + userId);
        if (Objects.nonNull(oldSession)) {
            if (oldSession.isOpen()) {
                oldSession.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.LOGIN_IN_OTHER_PLACE, "账户已在别处登录，请刷新")));
                oldSession.close();
            }
            return;
        }
        // 存储会话到map
        LIVE_ANCHOR_MAP.put(LiveConstant.LIVE_ANCHOR_PREFIX + roomId + "_" + userId, session);
        if (session.isOpen()) {
            session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok()));
        }
        if (liveRoom.getStartTime().compareTo(LocalDateTime.now()) < 0 && liveRoom.getEndTime().compareTo(LocalDateTime.now()) > 0
                && LiveRoomStatusType.isNeedOpen(liveRoom.getLiveStatus())) {
            // 更新直播为开播状态
            liveRoom.setLiveStatus(LiveRoomStatusType.LIVING.value());
            liveRoomService.updateById(liveRoom);
            // 发送消息通知用户直播状态变更
            sendChangeLiveRoomStatusMsg(roomId, LiveRoomStatusType.LIVING);
        }
    }

    @OnClose
    public void onClose(Session session, @PathParam("tenantId") String tenantId,
                        @PathParam(value = "userId") String userId, @PathParam(value = "roomId") String roomId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        logger.info("关闭主播直播连接");
        LIVE_ANCHOR_MAP.remove(LiveConstant.LIVE_ANCHOR_PREFIX + roomId + "_" + userId);
        setLiveRoomCloseStatus(userId, roomId);
    }

    private static void setLiveRoomCloseStatus(String userId, String roomId) throws IOException {
        LiveRoom liveRoom = liveRoomService.getOne(Wrappers.<LiveRoom>lambdaQuery()
                .ne(LiveRoom::getLiveStatus, LiveRoomStatusType.OFFLINE.value())
                .eq(LiveRoom::getId, roomId)
                .eq(LiveRoom::getUserId, userId)
                .gt(LiveRoom::getEndTime, LocalDateTime.now()));
        if (Objects.isNull(liveRoom)) {
            logger.error("主播当前没有直播");
            return;
        }
        if (liveRoom.getStartTime().compareTo(LocalDateTime.now()) < 0 && liveRoom.getEndTime().compareTo(LocalDateTime.now()) > 0
                && Objects.equals(liveRoom.getLiveStatus(), LiveRoomStatusType.LIVING.value())) {
            // 更新直播为暂停状态
            liveRoom.setLiveStatus(LiveRoomStatusType.STOP.value());
            liveRoomService.updateById(liveRoom);
            // 发送消息通知用户直播状态变更
            sendChangeLiveRoomStatusMsg(roomId, LiveRoomStatusType.STOP);
        }
        if (liveRoom.getEndTime().compareTo(LocalDateTime.now()) < 0
                && LiveRoomStatusType.isNeedFinish(liveRoom.getLiveStatus())) {
            // 更新直播为结束状态
            liveRoom.setLiveStatus(LiveRoomStatusType.FINISHED.value());
            liveRoomService.updateById(liveRoom);
            sendChangeLiveRoomStatusMsg(roomId, LiveRoomStatusType.FINISHED);
        }
    }

    private static void sendChangeLiveRoomStatusMsg(String roomId, LiveRoomStatusType liveRoomStatusType) throws IOException {
        // 发送消息通知用户直播状态变更
        List<Session> onlineUserList = LiveUserWebSocketServer.getOnlineUserList(roomId);
        LiveMsgItemVO liveMsgItem = new LiveMsgItemVO();
        liveMsgItem.setMsgType(MsgType.LIVE_ROOM_STATUS);
        liveMsgItem.setLiveStatus(liveRoomStatusType.value());
        if (CollUtil.isNotEmpty(onlineUserList)) {
            for (Session onlineUser : onlineUserList) {
                if (Objects.nonNull(onlineUser) && onlineUser.isOpen()) {
                    onlineUser.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(liveMsgItem)));
                }
            }
        }
    }

    @OnError
    public void onError(Session session, Throwable error, @PathParam("tenantId") String tenantId,
                        @PathParam(value = "userId") String userId, @PathParam("roomId") String roomId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        logger.error("主播直播连接发生错误{}", error.toString());
        LIVE_ANCHOR_MAP.remove(LiveConstant.LIVE_ANCHOR_PREFIX + roomId + "_" + userId);
        setLiveRoomCloseStatus(userId, roomId);
    }

    @OnMessage
    public void onMessage(String sourceMessage, Session session, @PathParam("tenantId") String tenantId, @PathParam(value = "userId") String userId,
                          @PathParam("roomId") String roomId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        if (StrUtil.isBlank(sourceMessage)) {
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CONTENT_FORMAT_ERROR, "主播消息内容格式有误")));
            }
            return;
        }
        logger.info("主播sourceMessage:" + sourceMessage);
        LiveMessageDTO liveMessageInfo = JSONUtil.toBean(sourceMessage, LiveMessageDTO.class);
        if (Objects.isNull(liveMessageInfo)) {
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CONTENT_FORMAT_ERROR, "主播消息内容格式有误")));
            }
            return;
        }
        if (checkLiveRoom(session, roomId, userId) == null) {
            return;
        }
        // 保持心跳
        if (Objects.equals(liveMessageInfo.getMsgType(), MsgType.HEART_BEAT)) {
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(MsgCode.HEART_BEAT, "保持心跳")));
            }
            return;
        }
        // 主播发送弹幕
        if (Objects.equals(liveMessageInfo.getMsgType(), MsgType.BULLET_SCREEN)) {
            if (StrUtil.isBlank(liveMessageInfo.getContent())) {
                return;
            }
            // 弹幕内容不超过50个字符
            if (liveMessageInfo.getContent().length() > LiveConstant.BULLET_SCREEN_MAX_LENGTH) {
                if (session.isOpen()) {
                    session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CONTENT_FORMAT_ERROR, "主播弹幕内容过长")));
                }
                return;
            }
            // 发送弹幕给在线用户
            LiveUserWebSocketServer.sendMsgToUser(roomId, liveMessageInfo.getNickName(), liveMessageInfo.getContent(), "1");
            return;
        }
        // 主播更新商品讲解状态
        if (Objects.equals(liveMessageInfo.getMsgType(), MsgType.EXPLAIN_STATUS)) {
            if (Objects.isNull(liveMessageInfo.getSpuId()) || Objects.isNull(liveMessageInfo.getExplainStatus())) {
                if (session.isOpen()) {
                    session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(MsgCode.CONTENT_FORMAT_ERROR, "主播消息内容格式有误")));
                }
                return;
            }
            if (Objects.equals(liveMessageInfo.getExplainStatus(), 0)) {
                // 商品取消讲解，发送消息
                sendNotExplainMsg(session, roomId, liveMessageInfo);
                return;
            }
            // 检查商品状态
            LiveRoomSpu liveRoomSpu = getLiveRoomSpu(roomId, liveMessageInfo);
            if (Objects.isNull(liveRoomSpu)) {
                // 没有该商品，发送商品状态异常消息给所有人
                sendRefreshProdMsg(session, roomId);
                return;
            }
            liveRoomSpu.setExplainStatus(liveMessageInfo.getExplainStatus());
            boolean updateStatus = liveRoomSpuService.update(liveRoomSpu, Wrappers.<LiveRoomSpu>lambdaQuery()
                    .eq(LiveRoomSpu::getSpuId, liveMessageInfo.getSpuId())
                    .eq(LiveRoomSpu::getRoomId, roomId));
            if (!updateStatus) {
                // 没有更新，发送商品状态异常消息给所有人
                sendRefreshProdMsg(session, roomId);
                return;
            }
            // 发送商品讲解状态更新消息给所有人
            sendChangeExplainStatusMsg(session, roomId, liveMessageInfo, liveRoomSpu);
        }
    }

    private static LiveRoomSpu getLiveRoomSpu(String roomId, LiveMessageDTO liveMessageInfo) {
        List<LiveRoomSpu> liveRoomSpuList = liveRoomSpuService.list(Wrappers.<LiveRoomSpu>lambdaQuery().eq(LiveRoomSpu::getRoomId, roomId));
        Map<String, LiveRoomSpu> liveProdMap = liveRoomSpuList.stream().collect(Collectors.toMap(LiveRoomSpu::getSpuId, p -> p));
        LiveRoomSpu liveRoomSpu = liveProdMap.get(liveMessageInfo.getSpuId());
        return liveRoomSpu;
    }

    private static void sendNotExplainMsg(Session session, String roomId, LiveMessageDTO liveMessageInfo) throws IOException {
        LiveRoomSpu liveRoomSpu = new LiveRoomSpu();
        liveRoomSpu.setExplainStatus(liveMessageInfo.getExplainStatus());
        liveRoomSpuService.update(liveRoomSpu, Wrappers.<LiveRoomSpu>lambdaQuery()
                .eq(LiveRoomSpu::getSpuId, liveMessageInfo.getSpuId())
                .eq(LiveRoomSpu::getRoomId, roomId));
        LiveMsgItemVO liveMsgItem = new LiveMsgItemVO();
        liveMsgItem.setMsgType(MsgType.EXPLAIN_STATUS);
        liveMsgItem.setSpuId(liveMessageInfo.getSpuId());
        liveMsgItem.setExplainStatus(liveMessageInfo.getExplainStatus());
        sendProdExplainStatusChangeMsg(session, roomId, liveMsgItem);
    }

    private static LiveRoom checkLiveRoom(Session session, String roomId, String userId) throws IOException {
        // 检查直播间
        LiveRoom liveRoom = liveRoomService.getOne(Wrappers.<LiveRoom>lambdaQuery()
                .ne(LiveRoom::getLiveStatus, LiveRoomStatusType.OFFLINE.value())
                .eq(LiveRoom::getId, roomId)
                .eq(LiveRoom::getUserId, userId)
                .gt(LiveRoom::getEndTime, LocalDateTime.now()));
        if (Objects.isNull(liveRoom)) {
            logger.error("主播当前没有直播");
            if (session.isOpen()) {
                // 当前未有直播
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CAN_NOT_FIND_ROOM_INFO, "主播当前没有直播")));
            }
        }
        return liveRoom;
    }

    private static void sendChangeExplainStatusMsg(Session session, String roomId, LiveMessageDTO liveMessageInfo, LiveRoomSpu liveRoomSpu) throws IOException {
        // 发送商品讲解状态更新消息给所有人AKAXtjlQNXc0t63VFy8exbN_f37UgelDnhmiETtSEdw1-nrJn_AGAoOK
        LiveMsgItemVO liveMsgItemVO = new LiveMsgItemVO();
        liveMsgItemVO.setMsgType(MsgType.EXPLAIN_STATUS);
        liveMsgItemVO.setSpuId(liveRoomSpu.getSpuId());
        liveMsgItemVO.setExplainStatus(liveMessageInfo.getExplainStatus());
        sendProdExplainStatusChangeMsg(session, roomId, liveMsgItemVO);
    }

    private static void sendProdExplainStatusChangeMsg(Session session, String roomId, LiveMsgItemVO liveMsgItem) throws IOException {
        if (session.isOpen()) {
            session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(liveMsgItem)));
        }
        // 给当前直播间在线用户也发送刷新商品列表的状态码
        List<Session> onlineUserList = LiveUserWebSocketServer.getOnlineUserList(roomId);
        // 给在线用户发送刷新商品列表消息
        if (CollUtil.isNotEmpty(onlineUserList)) {
            for (Session onlineUser : onlineUserList) {
                if (Objects.nonNull(onlineUser) && onlineUser.isOpen()) {
                    onlineUser.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(liveMsgItem)));
                }
            }
        }
    }

    private static void sendRefreshProdMsg(Session session, String roomId) throws IOException {
        // 发送刷新商品列表消息给所有人
        if (session.isOpen()) {
            session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(MsgCode.REFRUSH_PROD, "商品状态异常")));
        }
        // 给当前直播间在线用户也发送刷新商品列表的状态码
        List<Session> onlineUserList = LiveUserWebSocketServer.getOnlineUserList(roomId);
        // 给在线用户发送刷新商品列表消息
        if (CollUtil.isNotEmpty(onlineUserList)) {
            for (Session onlineUser : onlineUserList) {
                if (Objects.nonNull(onlineUser) && onlineUser.isOpen()) {
                    onlineUser.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(MsgCode.REFRUSH_PROD, "商品状态异常")));
                }
            }
        }
    }
}
